package com.winter.mq;

import com.winter.feign.DealServiceFeign;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

/**
 * @author JinHao
 * Date: 2023/04/18 17:00
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class RocketMqConsumer implements CommandLineRunner {

    public static final String NAME_SERVICE_ADDRESS = "localhost:9876";
    public static final String TOPIC = "JinHaoTopic";

    private final DealServiceFeign dealServiceFeign;

    @Override
    public void run(String... args) throws Exception {
        log.info("RocketMQ consumer 开始启动。。。");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        consumer.setNamesrvAddr(NAME_SERVICE_ADDRESS);
        consumer.setConsumerGroup("Consumer-Group");
        consumer.subscribe(TOPIC, "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            // 获取消息
            MessageExt messageExt = list.get(0);
            String tags = messageExt.getTags();
            String body = new String(messageExt.getBody());
            log.info("RocketMQ consumer 接收到消息 tag-{}-", tags);

            // 是否达到最大重试次数
            if (messageExt.getReconsumeTimes() == 6) {
                log.error("Receive message[msgId=" + messageExt.getMsgId() + "],[body=" + body + "]final failed!!!!");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }

            // 消息消费
            switch (tags) {
                case "test":
                    dealServiceFeign.dealTag(body);
                    break;
                case "":
                    break;
                default:
                    log.info("没有此 TAG-[" + tags + "]-对应的处理任务");
            }
            log.info("RocketMQ consumer 消费完消息 tag-{}-", tags);

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
        log.info("RocketMQ consumer 启动完成。。。");
    }
}
